package core.SparkStreaming;

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.streaming.StreamingQueryException;

import java.util.concurrent.TimeoutException;

public class SparkKafkaConsumer {
    public static void main(String[] args) throws StreamingQueryException, TimeoutException {
        // 创建 SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("SparkKafkaConsumer")
                .master("local[*]") // 使用本地模式运行
                .getOrCreate();

        // 设置日志级别为 ERROR，避免过多日志输出
        spark.sparkContext().setLogLevel("ERROR");

        // 从 Kafka 读取数据
        Dataset<Row> df = spark.readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "114.251.235.19:9092") // Kafka broker 地址
                .option("subscribe", "topic_test") // 订阅的 topic
                .load();

        // 将二进制数据转换为字符串
        Dataset<Row> valueDf = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)");

        // 打印接收到的消息
        valueDf.writeStream()
                .outputMode(OutputMode.Append())
                .format("console") // 输出到控制台
                .start()
                .awaitTermination(); // 等待流处理结束
    }
}
